package com.shujia.wyh.jinjie;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import redis.clients.jedis.Jedis;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

public class HbaseAndRedisDemo {
    private static Jedis jedis;
    private static Connection conn;
    private static Admin admin;

    public static void main(String[] args) throws Exception {
        //获取与redis的连接对象
        jedis = new Jedis("192.168.254.100", 7000);
        jedis.select(1); //选择1号

        //获取hbase的连接对象
        Configuration conf = new Configuration();
        //2、设置以下zookeeper的集群信息
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();

        System.out.println("获取连接成功！！" + conn);
        System.out.println("HMaster的对象获取成功！！" + admin);
        System.out.println("Redis的对象获取成功！！" + jedis);
        System.out.println("------------------------------------------------------------------");

        //第一步：读取hbase中的数据，将男生和女生的学号分别取出来
        TableName students = TableName.valueOf("students");

        Table table = conn.getTable(students);

//        List<String> boyIds = getBoyIds(table);
//        List<String> girlIds = getGirlIds(table);
//
//        //将结果放入到redis
//        for (String boyId : boyIds) {
//            jedis.sadd("bigdata25:男",boyId);
//        }
//        for (String girlId : girlIds) {
//            jedis.sadd("bigdata25:女",girlId);
//        }
//
//        System.out.println("索引构建成功！！");

        getData(table);


        //释放资源
        admin.close();
        conn.close();
        jedis.close();
    }

    public static void getData(Table table) throws Exception {
        //反向通过redis查询学号，在去hbase中查询其他列数据
        Set<String> boySet = jedis.smembers("bigdata25:男");
        Set<String> girlSet = jedis.smembers("bigdata25:女");

        System.out.println("===========================男生前10条数据=======================================");
        printData(boySet,table);
        System.out.println("===========================女生前10条数据=======================================");
        printData(girlSet,table);

    }

    public static void printData(Set<String> set, Table table) throws Exception {
        int i = 0;
        for (String id : set) {
            if(i==10){
                break;
            }
            Get get = new Get(Bytes.toBytes(id));
            Result result = table.get(get);
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            System.out.println("学号：" + id + ", 姓名：" + name + ", 年龄" + age + ", 性别：" + gender + ", 班级：" + clazz);
            i++;
        }
    }


    public static List<String> getBoyIds(Table table) throws Exception {
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("info"),
                Bytes.toBytes("gender"),
                CompareOperator.EQUAL,
                new BinaryPrefixComparator(Bytes.toBytes("男"))
        );
        Scan scan = new Scan();
        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);
        //创建一个集合，存储学号
        ArrayList<String> ids = new ArrayList<>();

        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());
            ids.add(id);
        }

        return ids;

    }

    public static List<String> getGirlIds(Table table) throws Exception {
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("info"),
                Bytes.toBytes("gender"),
                CompareOperator.EQUAL,
                new BinaryPrefixComparator(Bytes.toBytes("女"))
        );
        Scan scan = new Scan();
        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);
        //创建一个集合，存储学号
        ArrayList<String> ids = new ArrayList<>();

        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());
            ids.add(id);
        }

        return ids;
    }
}
